package iot.hbase.codec;

import hbase.template.CellMapper;
import hbase.template.RowMapper;
import hbase.template.RowSerializer;
import iot.entity.Registration;
import iot.utils.DataPackUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * Created by dave on 18-7-21 上午9:51.
 */
public class RegistrationCodec implements RowSerializer<Registration>, RowMapper<Registration> {

    @Override
    public Registration mapRow(Result result) {
        Registration r = new Registration();
        byte[] f = "r".getBytes();
        r.setSim(CellMapper.toString(result, "r", "si"));
        r.setProvinceId(CellMapper.toInt(result, "r", "pid"));
        r.setCityId(CellMapper.toInt(result, "r", "cid"));
        r.setMakerId(CellMapper.toString(result, "r", "mid"));
        r.setModel(CellMapper.toString(result, "r", "mo"));
        r.setDeviceId(CellMapper.toString(result, "r", "did"));
        r.setColorId(CellMapper.toInt(result, "r", "co"));

        Cell c = result.getColumnLatestCell(f, "vin".getBytes());
        if (c != null) {
            r.setVin(Bytes.toString(CellUtil.cloneValue(c)));
        } else {
            c = result.getColumnLatestCell(f, "lic".getBytes());
            r.setLicense(Bytes.toString(CellUtil.cloneValue(c)));
        }
        r.setTime(new Date(CellMapper.toLong(result, "r", "tm")));
        return r;
    }

    @Override
    public List<KeyValue> serialize(Registration record) {
        List<KeyValue> kv = new ArrayList<>();
        byte[] key = generateRowKey(record);
        byte[] family = "r".getBytes();
        kv.add(new KeyValue(key, family, "si".getBytes(), Bytes.toBytes(record.getSim())));
        kv.add(new KeyValue(key, family, "pid".getBytes(), Bytes.toBytes(record.getProvinceId())));
        kv.add(new KeyValue(key, family, "cid".getBytes(), Bytes.toBytes(record.getCityId())));
        kv.add(new KeyValue(key, family, "mid".getBytes(), Bytes.toBytes(record.getMakerId())));
        kv.add(new KeyValue(key, family, "mo".getBytes(), Bytes.toBytes(record.getModel())));
        kv.add(new KeyValue(key, family, "did".getBytes(), Bytes.toBytes(record.getDeviceId())));
        kv.add(new KeyValue(key, family, "co".getBytes(), Bytes.toBytes(record.getColorId())));
        if (record.getVin() != null) {
            kv.add(new KeyValue(key, family, "vin".getBytes(), Bytes.toBytes(record.getVin())));
        }
        if (record.getLicense() != null) {
            kv.add(new KeyValue(key, family, "lic".getBytes(), Bytes.toBytes(record.getLicense())));
        }
        kv.add(new KeyValue(key, family, "tm".getBytes(), Bytes.toBytes(record.getTime().getTime())));
        return kv;
    }

    @Override
    public byte[] generateRowKey(Registration record) {
        byte[] id = Bytes.toBytes(record.getSim());
        byte[] bytes = new byte[id.length + 8];
        System.arraycopy(DataPackUtil.getBytesFromUUID(record.getUuid()), 0, bytes, 0, 8);
        System.arraycopy(id, 0, bytes, 8, id.length);
        return bytes;
    }
}
